package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.flink.utils.HbaseSinkFunction;
import net.bwie.flink.utils.MysqlCdcUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 将MySQL数据库gmall中维表数据，实时同步到HBase表中
 * @author xuanyu
 * @date 2025/10/31
 */
public class _01SyncDimData {

	public static void main(String[] args) throws Exception{
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2. 数据源-source
		DataStream<String> stream = MysqlCdcUtil.cdcMysql(env, "gmall", "base_category3");
		//stream.print("cdc");

		// 3. 数据转换-transformation
		SingleOutputStreamOperator<String> stream1 = stream.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					// 解析JSON
					Object afterValue = JSON.parseObject(value).get("after");
					// 判断
					return null != afterValue;
				}
			}
		);

		// todo 向HBase表写入数据时，自定义Sink要求数据类型为二元组，Key为RowKey主键，Value为json字符串。
		SingleOutputStreamOperator<Tuple2<String, String>> stream2 = stream1.map(
			new MapFunction<String, Tuple2<String, String>>() {
				@Override
				public Tuple2<String, String> map(String value) throws Exception {
					// 解析jsonoperate_time
					JSONObject jsonObject = JSON.parseObject(value).getJSONObject("after");
					// 获取主键dic_code
					String rowKeyValue = jsonObject.getString("id");
					// 返回
					return Tuple2.of(rowKeyValue, jsonObject.toJSONString());
				}
			}
		);

		// 4. 数据接收器-sink
		HbaseSinkFunction sink = new HbaseSinkFunction(
			"dim_base_category3", "info", "name,category2_id"
		) ;
		// 先在HBase中创建表 -> create 'dim_base_category3', 'info'
		stream2.addSink(sink);

		// 5. 触发执行-execute
		env.execute("SyncDimData");
	}

}
